package com.lwx.mq.kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import kafka.admin.AdminUtils;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.utils.ZKStringSerializer$;

import org.I0Itec.zkclient.ZkClient;


public class KafkaProducer extends Thread {
	public static void main(String[] args) {
		String topicName = "topictest";
		Producer producer = createProducer("test1:9092", "test1:2181/kafkatest",topicName, 1, 1);
		producer.send(new KeyedMessage<String, String>(topicName, "taa"));
		producer.close();
	}
	
	public static Producer<String, String> createProducer(String brokerList, String zkConnect, String topicName, int numPartitions, int replicationFactor) {
		createTopic(zkConnect, topicName, numPartitions, replicationFactor);
		Properties props = new Properties();
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		props.put("metadata.broker.list", brokerList);
		props.put("request.required.acks", "1");
		ProducerConfig config = new ProducerConfig(props);
		return new Producer<String, String>(config);
	}
	
	public static boolean createTopic(String zkConnect, String topicName, int numPartitions, int replicationFactor){
		ZkClient zkClient = null;
		try {
			zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);
			boolean isTopicExist = AdminUtils.topicExists(zkClient, topicName);
			if(!isTopicExist){
				Properties topicConfig = new Properties();
				AdminUtils.createTopic(zkClient, topicName, numPartitions, replicationFactor, topicConfig);
			}
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally{
			if(zkClient!=null){
				zkClient.close();
			}
		}
	}
}